/*-------------------------------------------------------------------------
 *
 * polar_xlog_idx.c
 *   Implementation of parse xlog records.
 *
 * Portions Copyright (c) 2021, Alibaba Group Holding limited
 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *   src/backend/access/logindex/polar_xlog_idx.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/polar_logindex_redo.h"
#include "catalog/pg_control.h"
#include "miscadmin.h"
#include "storage/buf_internals.h"
#include "access/polar_logindex_redo.h"

static XLogRedoAction
xlog_idx_fpi_redo(XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	XLogRedoAction action = BLK_NOTFOUND;
	uint8		block_id;

	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
	{
		BufferTag	page_tag;

		POLAR_GET_LOG_TAG(record, page_tag, block_id);

		if (BUFFERTAGS_EQUAL(*tag, page_tag))
		{
			/*
			 * XLOG_FPI records contain nothing else but one or more block
			 * references. Every block reference must include a full-page
			 * image even if full_page_writes was disabled when the record was
			 * generated - otherwise there would be no point in this record.
			 *
			 * XLOG_FPI_FOR_HINT records are generated when a page needs to be
			 * WAL-logged because of a hint bit update. They are only
			 * generated when checksums and/or wal_log_hints are enabled. They
			 * may include no full-page images if full_page_writes was
			 * disabled when they were generated. In this case there is
			 * nothing to do here.
			 *
			 * No recovery conflicts are generated by these generic records -
			 * if a resource manager needs to generate conflicts, it has to
			 * define a separate WAL record type and redo routine.
			 */
			if (!XLogRecHasBlockImage(record, block_id))
			{
				if (XLogRecGetInfo(record) == XLOG_FPI)
					elog(ERROR, "XLOG_FPI record did not contain a full-page image");
			}
			else
				action = POLAR_READ_BUFFER_FOR_REDO(record, block_id, buffer);
			break;
		}
	}

	return action;
}

/*
 * POLAR: fullpage snapshot image wal redo
 */
static bool
xlog_idx_fpsi_redo(polar_logindex_redo_ctl_t instance, XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	XLogRedoAction action = BLK_NOTFOUND;
	BufferTag	tag0;
	Page		page;
	uint64		fullpage_no = 0;

	POLAR_GET_LOG_TAG(record, tag0, 0);

	if (BUFFERTAGS_EQUAL(*tag, tag0))
	{
		page = BufferGetPage(*buffer);
		/* get fullpage_no from record */
		memcpy(&fullpage_no, XLogRecGetData(record) + sizeof(PolarWalType), sizeof(uint64));
		/* read fullpage from file */
		polar_read_fullpage(instance->fullpage_ctl, page, fullpage_no);
		action = BLK_RESTORED;
	}

	return action;
}

void
polar_xlog_idx_save(polar_logindex_redo_ctl_t instance, XLogReaderState *record)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
	uint8		block_id;

	switch (info)
	{
		case XLOG_FPI:
		case XLOG_FPI_FOR_HINT:
			for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
			{
				if (!XLogRecHasBlockImage(record, block_id))
				{
					if (info == XLOG_FPI)
						elog(ERROR, "XLOG_FPI record did not contain a full-page image");
					continue;
				}
				polar_logindex_save_block(instance, record, block_id);
			}
			break;
		case POLAR_WAL:
			{
				if (*((PolarWalType *) XLogRecGetData(record)) == PWT_FPSI)
				{
					for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
						polar_fullpage_logindex_save_block(instance, record, block_id);
				}
			}
			break;

		default:
			break;
	}
}

bool
polar_xlog_idx_parse(polar_logindex_redo_ctl_t instance, XLogReaderState *record)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
	uint8		block_id;

	if (info == XLOG_FPI || info == XLOG_FPI_FOR_HINT)
	{
		for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
		{
			if (!XLogRecHasBlockImage(record, block_id))
			{
				if (info == XLOG_FPI)
					elog(ERROR, "XLOG_FPI record did not contain a full-page image");
				continue;
			}
			polar_logindex_redo_parse(instance, record, block_id);
		}
		return true;
	}
	else if (info == POLAR_WAL && *((PolarWalType *) XLogRecGetData(record)) == PWT_FPSI)
	{
		uint64		fullpage_no = 0;

		/* get fullpage_no from record */
		memcpy(&fullpage_no, XLogRecGetData(record) + sizeof(PolarWalType), sizeof(uint64));
		/* Update logindex max_fullpage_no */
		polar_update_max_fullpage_no(instance->fullpage_ctl, fullpage_no);
		return true;
	}

	return false;
}

XLogRedoAction
polar_xlog_idx_redo(polar_logindex_redo_ctl_t instance, XLogReaderState *record, BufferTag *tag, Buffer *buffer)
{
	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

	/*
	 * These operations don't overwrite MVCC data so no conflict processing is
	 * required. The ones in heap2 rmgr do.
	 */

	switch (info)
	{
		case XLOG_FPI:
		case XLOG_FPI_FOR_HINT:
			return xlog_idx_fpi_redo(record, tag, buffer);
		case POLAR_WAL:
			{
				if (*((PolarWalType *) XLogRecGetData(record)) == PWT_FPSI)
					return xlog_idx_fpsi_redo(instance, record, tag, buffer);
			}
			break;

		default:
			elog(PANIC, "polar_xlog_idx_redo: unknown op code %u", info);
			break;
	}

	return BLK_NOTFOUND;
}
